package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/*
4）、（DIM层维度表）编写Flink 流式程序，实时消费ODS层物流业务数据，将核心维表数据（用户信息表user_info、字典表base_dic、地区表base_region_info、快递员信息表（、express_courier）不同类型数据，并对json数据解析过滤，实时存储业务数据到HBase表（表名称前缀：dim_，表列簇名称：info）。（5分）
5）、（DWD层事实表）编写Flink 流式程序，实时消费ODS层物流业务数据，过滤获取运单表（order_info）数据，按照运单状态（status）将运单数据划分不同类型数据，使用侧边流sideOutput方式，将不同状态运单数据存储Kafka不同Topic主题；（4分）
 */
public class Test3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //（3）指定流处理环境以及检查点相关的设置，从kafka的tms_ods主题中读取，创建消费者对象，消费数据  封装为流，筛选订单和订单明细数据，
        // 对流中的数据类型进行转换 ，
        // 按照order_id进行分组，定义侧输出流标签 下单放到主流，支付成功、取消运单、揽收(接单)、发单 转运完成、派送成功、签收放到侧输出流，
        // 从主流中提取侧输出流，将不同流的数据写到kafka的不同主题中（5分）
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "test662");
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("tms_ods_yk6", new SimpleStringSchema(), properties));
        stream.print();
        //{database: data:  type: table:}

        SingleOutputStreamOperator<String> oiDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject.getString("table").equals("order_info")) {
                        return true;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return false;
            }
        });
//        SingleOutputStreamOperator<String> oiDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));

        // 订单详情基本没用
        SingleOutputStreamOperator<String> ocDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));

        // 订单表转成JSONobject
        SingleOutputStreamOperator<JSONObject> oiMapDS = oiDS.map(x -> JSON.parseObject(x));

        OutputTag<String> t60020 = new OutputTag<String>("t60020"){};
        //60030
        OutputTag<String> t60030 = new OutputTag<String>("t60030"){};
        //60050
        OutputTag<String> t60050 = new OutputTag<String>("t60050"){};
        //60060
        OutputTag<String> t60060 = new OutputTag<String>("t60060"){};
        //60070
        OutputTag<String> t60070 = new OutputTag<String>("t60070"){};
        //60080
        OutputTag<String> t60080 = new OutputTag<String>("t60080"){};
        //60999
        OutputTag<String> t60999 = new OutputTag<String>("t60999"){};
        SingleOutputStreamOperator<String> t60010DS = oiMapDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject s, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //{"database":"tms01","data":{},"type":"insert","table":"order_info"}
                JSONObject js = s.getJSONObject("data");
                String stauts = js.getString("status");
                if ("60010".equals(stauts)) {
                    collector.collect(js.toString());
                } else if ("60020".equals(stauts)) {
                    context.output(t60020, js.toString());
                } else if ("60030".equals(stauts)) {
                    context.output(t60030, js.toString());
                } else if ("60050".equals(stauts)) {
                    context.output(t60050, js.toString());
                } else if ("60060".equals(stauts)) {
                    context.output(t60060, js.toString());
                } else if ("60070".equals(stauts)) {
                    context.output(t60070, js.toString());
                } else if ("60080".equals(stauts)) {
                    context.output(t60080, js.toString());
                } else if ("60999".equals(stauts)) {
                    context.output(t60999, js.toString());
                }
            }
        });
        FlinkKafkaProducer<String> p60010 = new FlinkKafkaProducer<>("dwd-order-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60020 = new FlinkKafkaProducer<>("dwd-pay-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60030 = new FlinkKafkaProducer<>("dwd-receive-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60050 = new FlinkKafkaProducer<>("dwd-dispatch-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60060 = new FlinkKafkaProducer<>("dwd-bound-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60070 = new FlinkKafkaProducer<>("dwd-deliver-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60080 = new FlinkKafkaProducer<>("dwd-sign-detail_yk6", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60999 = new FlinkKafkaProducer<>("dwd-cancel-detail_yk6", new SimpleStringSchema(), properties);
        // 主流
        t60010DS.addSink(p60010);
        t60010DS.getSideOutput(t60020).addSink(p60020);
        t60010DS.getSideOutput(t60030).addSink(p60030);
        t60010DS.getSideOutput(t60050).addSink(p60050);
        t60010DS.getSideOutput(t60060).addSink(p60060);
        t60010DS.getSideOutput(t60070).addSink(p60070);
        t60010DS.getSideOutput(t60080).addSink(p60080);
        t60010DS.getSideOutput(t60999).addSink(p60999);

        env.execute();
    }
}
